package com.atguigu.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.base.BaseApp;
import com.atguigu.gmall.realtime.common.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.DateFormatUtil;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

public class DwsTradeSkuOrderWindowSyncCache extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeSkuOrderWindowSyncCache().start(10029, 4, "dws_trade_sku_order_window", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        // TODO 1.对流中数据进行类型转换并过滤空消息 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                            collector.collect(jsonObj);
                        }
                    }
                }
        );
        //jsonObjDS.print();
        // TODO 2.按照唯一键（订单明细id）进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getString("id"));
        // TODO 3.去重
        // 3.1 状态 + 定时器 缺点：时效性差，不管是否重复都要等待10s
        /*SingleOutputStreamOperator<JSONObject> jsonObjDS1 = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj == null) {
                            // 说明目前没有重复，将当前数据放到状态中去，并注册10s后执行的定时器
                            lastJsonObjState.update(jsonObject);
                            TimerService timerService = context.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 10000);
                        } else {
                            //说明重复了，将当前数据的聚合时间和状态中存在的数据聚合时间进行比较
                            String ts1 = jsonObject.getString("聚合时间");
                            String ts2 = lastJsonObj.getString("聚合时间");
                            if (ts1.compareTo(ts2) > 0) {
                                lastJsonObjState.update(jsonObject);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = lastJsonObjState.value();
                        out.collect(jsonObj);
                    }
                }
        );*/
        // 3.2 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    ValueState<JSONObject> lastJsonObjState;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    //{"create_time":"2024-08-03 15:35:26","sku_num":"1",
                    // "split_original_amount":"9197.0000",
                    // "split_coupon_amount":"0.0","sku_id":"12","date_id":"2024-08-03","user_id":"1643","province_id":"24","sku_name":"Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机","id":"8550","order_id":"6071","split_activity_amount":"0.0","split_total_amount":"9197.0","ts":1654760126}
                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        // 从状态中获取数据
                        JSONObject lastJsonObj = lastJsonObjState.value();
                        if (lastJsonObj != null) {
                            // 重复了 将状态中影响到度量值的字段进行取反，在向下游传递
                            String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                            String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                            String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                            String splitTotalAmount = lastJsonObj.getString("split_total_amount");

                            lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                            lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                            lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                            lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);

                            collector.collect(lastJsonObj);
                        }
                        lastJsonObjState.update(jsonObject);
                        collector.collect(jsonObject);
                    }
                }
        );
        //distinctDS.print();
        // TODO 4.再次对流中数据进行类型转换 jsonObj -> 统计的实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean map(JSONObject jsonObject) throws Exception {
                        return TradeSkuOrderBean.builder()
                                .skuId(jsonObject.getString("sku_id"))
                                .originalAmount(jsonObject.getBigDecimal("split_original_amount"))
                                .couponReduceAmount(jsonObject.getBigDecimal("split_coupon_amount"))
                                .activityReduceAmount(jsonObject.getBigDecimal("split_activity_amount"))
                                .orderAmount(jsonObject.getBigDecimal("split_total_amount"))
                                .ts(jsonObject.getLong("ts") * 1000)
                                .build();
                    }
                }
        );
        // TODO 5.指定WaterMark的生成策略以及提取时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeSkuOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeSkuOrderBean tradeSkuOrderBean, long l) {
                                        return tradeSkuOrderBean.getTs();
                                    }
                                }
                        )
        );
        // TODO 6.按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuKeyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);
        // TODO 7.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS = skuKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        // TODO 8.聚合
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSkuOrderBean>() {
                    @Override
                    public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                        value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.setActivityReduceAmount(value1.getActivityReduceAmount().add(value2.getActivityReduceAmount()));
                        value1.setCouponReduceAmount(value1.getCouponReduceAmount().add(value2.getCouponReduceAmount()));
                        return value1;
                    }
                },
                new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String value, TimeWindow timeWindow, Iterable<TradeSkuOrderBean> iterable, Collector<TradeSkuOrderBean> collector) throws Exception {
                        TradeSkuOrderBean tradeSkuOrderBean = iterable.iterator().next();
                        tradeSkuOrderBean.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        tradeSkuOrderBean.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        tradeSkuOrderBean.setCurDate(DateFormatUtil.tsToDateForPartition(timeWindow.getEnd()));
                        collector.collect(tradeSkuOrderBean);
                    }
                }
        );
        reduceDS.print();//度量值无负值（测试）
        // TODO 9.关联sku维度
        // 维度关联最基本的方式 缺点频繁交互
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    Connection hBaseConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseConnection);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                        //根据流中对象获取要关联的维度的主键
                        String skuId = tradeSkuOrderBean.getSkuId();
                        //根据维度主键到维度表中获取维度对象
                        JSONObject dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                        //将维度对象的相关属性补充到流中对象上 id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        tradeSkuOrderBean.builder()
                                .spuId(dimJsonObj.getString("spu_id"))
                                .spuName(dimJsonObj.getString("sku_name"))
                                .trademarkId(dimJsonObj.getString("tm_id"))
                                .category3Id(dimJsonObj.getString("category3_id"))
                                .build();
                        return tradeSkuOrderBean;
                    }
                }
        );*/
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {

                    private Jedis jedis;
                    private Connection hBaseConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseConnection);
                        RedisUtil.closeJedis(jedis);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                        // 根据流中对象获取要关联的维度主键
                        String skuId = tradeSkuOrderBean.getSkuId();
                        // 先从redis缓存中好去维度数据
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_sku_info", skuId);
                        if (dimJsonObj != null){
                            // 如果获取到，直接将其作为返回值返回（缓存命中）
                            System.out.println("从Redis中获得dim_sku_info的维度数据");
                        } else {
                            // 如果没有获取到，发送请求到HBase中查询维度
                            dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, "dim_sku_info", skuId, JSONObject.class);
                            if (dimJsonObj != null){
                                System.out.println("从HBase中获得dim_sku_info的维度数据");
                                // 将查询结果放到Redis
                                RedisUtil.writeDim(jedis,"dim_sku_info", skuId, dimJsonObj);
                            } else {
                                System.out.println("HBase中没有dim_sku_info的维度数据");
                            }
                        }

                        if (dimJsonObj != null){
                            tradeSkuOrderBean.builder()
                                    .spuId(dimJsonObj.getString("spu_id"))
                                    .spuName(dimJsonObj.getString("sku_name"))
                                    .trademarkId(dimJsonObj.getString("tm_id"))
                                    .category3Id(dimJsonObj.getString("category3_id"))
                                    .build();
                        }
                        // 将查询出来的维度数据补充到流对象上
                        return tradeSkuOrderBean;
                    }
                }
        );
        withSkuInfoDS.print();
        // TODO 10.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = withSkuInfoDS.map(
                new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                    private Jedis jedis;
                    private Connection hBaseConnection;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hBaseConnection = HBaseUtil.getHBaseConnection();
                        jedis = RedisUtil.getJedis();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hBaseConnection);
                        RedisUtil.closeJedis(jedis);
                    }

                    @Override
                    public TradeSkuOrderBean map(TradeSkuOrderBean tradeSkuOrderBean) throws Exception {
                        // 根据流中对象获取要关联的维度主键
                        String spuId = tradeSkuOrderBean.getSpuId();
                        // 先从redis缓存中好去维度数据
                        JSONObject dimJsonObj = RedisUtil.readDim(jedis, "dim_spu_info", spuId);
                        if (dimJsonObj != null) {
                            // 如果获取到，直接将其作为返回值返回（缓存命中）
                            System.out.println("从Redis中获得dim_spu_info的维度数据");
                        } else {
                            // 如果没有获取到，发送请求到HBase中查询维度
                            dimJsonObj = HBaseUtil.getRow(hBaseConnection, Constant.HBASE_NAMESPACE, "dim_spu_info", spuId, JSONObject.class);
                            if (dimJsonObj != null) {
                                System.out.println("从HBase中获得dim_spu_info的维度数据");
                                // 将查询结果放到Redis
                                RedisUtil.writeDim(jedis, "dim_spu_info", spuId, dimJsonObj);
                            } else {
                                System.out.println("HBase中没有dim_spu_info的维度数据");
                            }
                        }

                        if (dimJsonObj != null) {
                            // 将查询出来的维度数据补充到流对象上
                            // id,spu_name,description,category3_id,tm_id
                            tradeSkuOrderBean.builder()
                                    .spuName(dimJsonObj.getString("spu_name"))
                                    .spuName(dimJsonObj.getString("sku_name"))
                                    .trademarkId(dimJsonObj.getString("tm_id"))
                                    .category3Id(dimJsonObj.getString("category3_id"))
                                    .build();
                        }
                        return tradeSkuOrderBean;
                    }
                }
        );
        // TODO 11.关联品牌（tm）维度
        // TODO 12.关联category3维度
        // TODO 13.关联category2维度
        // TODO 14.关联category1维度
        // TODO 15.关联结果写入doris
    }
}
